package com.google.bitcoin.db;


import com.google.bitcoin.core.*;
import com.google.bitcoin.core.TransactionConfidence.ConfidenceType;

import java.util.List;
import java.util.LinkedList;
import java.util.Collection;
import java.util.HashSet;
import java.util.HashMap;
import java.util.Set;
import java.util.TreeSet;
import java.util.Map;
import java.util.TreeMap;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;

import java.math.BigInteger;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.services.dynamodb.AmazonDynamoDBClient;

import com.amazonaws.services.dynamodb.model.DeleteItemRequest;
import com.amazonaws.services.dynamodb.model.GetItemRequest;
import com.amazonaws.services.dynamodb.model.GetItemResult;
import com.amazonaws.services.dynamodb.model.Key;
import com.amazonaws.services.dynamodb.model.PutItemRequest;
import com.amazonaws.services.dynamodb.model.UpdateItemRequest;
import com.amazonaws.services.dynamodb.model.AttributeAction;
import com.amazonaws.services.dynamodb.model.AttributeValue;
import com.amazonaws.services.dynamodb.model.AttributeValueUpdate;
import com.amazonaws.services.dynamodb.model.ExpectedAttributeValue;
import com.amazonaws.services.dynamodb.model.QueryRequest;
import com.amazonaws.services.dynamodb.model.QueryResult;
import com.amazonaws.services.dynamodb.model.ScanRequest;
import com.amazonaws.services.dynamodb.model.ScanResult;

import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;

import org.apache.commons.codec.binary.Base64;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.Semaphore;


public class TxDynamoDB extends TransactionDB
{
    private static final Logger log = LoggerFactory.getLogger(TxDynamoDB.class);
    private AmazonDynamoDBClient dynamo;
    private AmazonS3Client s3;
    private final String table_tx;
    private final String table_unspent;
    private final String s3_bucket;
    private boolean consistent_read=true;
    private NetworkParameters params;
    private BlockChainCache block_chain_cache;

    private HashMap<Sha256Hash, Map<String, AttributeValue> > stupid_cache;
    private HashMap<String, Map<String, AttributeValue> > stupid_unspent;

    private LRUCache<Sha256Hash, Map<String, AttributeValue> > tx_cache;

    private static final int HOPELESS_HOURS=12;

    public class TX
    {
        public static final String ID="txid";
        public static final String FIRST_SEEN="first_seen";
        public static final String BLOCK_LIST="block_list";
        public static final String SPENT_BY="spent_by";
        public static final String BYTES="bytes";
        public static final String S3LOC="s3loc";

    }

        public static Key TXgetKey(Sha256Hash tx_id)
        {
            return TXgetKey(tx_id.toString());
        }
        public static Key TXgetKey(String tx_id)
        {
            return new Key(new AttributeValue(tx_id));
        }
        public static String TXgetSpentKey(long idx)
        {
            return "out:" + idx;
        }
 
    public class UNSPENT
    {
        public static final String ADDRESS="address";
        public static final String OUTPOINT="outpoint";
        public static final String VALUE="value";

    }
        public static Key UNSPENTgetKey(Address addr, Sha256Hash tx, long out_idx)
        {
            return UNSPENTgetKey(addr, tx.toString(), out_idx);
        }
        public static Key UNSPENTgetKey(Address addr, String tx, long out_idx)
        {
            return new Key(new AttributeValue(addr.toString()), new AttributeValue(tx +":" + out_idx));
        }
        public static String UNSPENTgetOutpoint(Sha256Hash tx, long out_idx)
        {
            return tx.toString() +":" + out_idx;
        }


    public TxDynamoDB(NetworkParameters params, AWSCredentials creds, String endpoint, String table_base, BlockChainCache block_chain_cache)
    {   
        log.debug("I have a debug");
        dynamo = new DynamoRateLimited(creds, 100.0);
        dynamo.setEndpoint(endpoint);

        s3 = new AmazonS3Client(creds);

        table_tx = table_base +"-transactions"; 
        table_unspent = table_base +"-unspent"; 
        this.params = params;

        s3_bucket = "txdynamodb-" + table_base + "-transactions";

        this.block_chain_cache = block_chain_cache;

        tx_cache=new LRUCache<Sha256Hash, Map<String, AttributeValue>>(10240);

        
    }

    public void invalidateCache(Sha256Hash hash)
    {
        synchronized(tx_cache)
        {
            tx_cache.remove(hash);
        }
    }

    public Transaction getTransaction(Sha256Hash hash)
    {
        return getTransaction(hash, false);
    }
    public Transaction getTransaction(Sha256Hash hash, boolean cache_ok)
    {
        Map<String, AttributeValue> map = getTransactionMap(hash, cache_ok);
        if (map==null) return null;
        return decodeTransaction(map); 
    }
    public Transaction getTransaction(Sha256Hash hash, int connected_out_levels)
    {

        if (connected_out_levels==0) return getTransaction(hash, true);

        Map<String, AttributeValue> map = getTransactionMap(hash, true);
        Transaction tx = decodeTransaction(map);

        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            
            AttributeValue av = map.get(TXgetSpentKey(idx));
            if (av != null)
            {
                LinkedList<String> lst = new LinkedList<String>();
                lst.addAll(av.getSS());
                //just taking first because whatever
                String child_hash = lst.getFirst();

                Transaction child = getTransaction(new Sha256Hash(child_hash), connected_out_levels-1);
                for(TransactionInput in : child.getInputs())
                {

                    if ((in.getOutpoint().getHash().equals(hash)) && (in.getOutpoint().getIndex()==idx))
                    {
                        out.markAsSpent(in);
                    }
                }

            }
            idx++;
        }

        return tx;
    }

    public Map<String, AttributeValue> getTransactionMap(Sha256Hash hash)
    {
        return getTransactionMap(hash, false);
    }
    public Map<String, AttributeValue> getTransactionMap(Sha256Hash hash, boolean cache_ok)
    {
        if (stupid_cache!=null)
        {
            if (stupid_cache.containsKey(hash)) return stupid_cache.get(hash);
        }
        if (cache_ok)
        {
            synchronized(tx_cache)
            {
                if (tx_cache.containsKey(hash)) return tx_cache.get(hash);
            }
        }
        log.debug("Going to db for: " + hash);
        
        GetItemRequest get = new GetItemRequest(table_tx, TXgetKey(hash));
        get.setConsistentRead(consistent_read);

        try
        {
            GetItemResult res = dynamo.getItem(get);
            Map<String, AttributeValue> item =  res.getItem();
            if (stupid_cache != null)
            {
                stupid_cache.put(hash, item);
            }
            synchronized(tx_cache)
            {
                tx_cache.put(hash,item);
            }
            return item;

        }
        catch(com.amazonaws.services.dynamodb.model.ResourceNotFoundException e)
        {
            if (stupid_cache != null)
            {
                stupid_cache.put(hash, null);
            }

            return null;
        }
    }

    @Override
    public BigInteger getBalance(final BalanceType balanceType, HashSet<Address> addresses)
    {
        final Semaphore wait_sem=new Semaphore(0);
        final AtomicLong total = new AtomicLong(0);
        int count=addresses.size();

        for(final Address a : addresses)
        {
            executor.execute(new Runnable(){

                public void run()
                {
                    BigInteger value = getBalance(balanceType, a);

                    total.addAndGet(value.longValue());

                    wait_sem.release();

                }

            });
        }
        wait_sem.acquireUninterruptibly(count);

        return BigInteger.valueOf(total.get());   
    }

    @Override
    public BigInteger getBalance(BalanceType balanceType, Address addr)
    {
        if (balanceType!=BalanceType.ESTIMATED)
        {
            return super.getBalance(balanceType, addr);
        }
        BigInteger total = BigInteger.ZERO;

        QueryRequest query = new QueryRequest(table_unspent, new AttributeValue(addr.toString()));
        query.setLimit(100);

        Key last_key=null;

        do
        {
            QueryResult res = dynamo.query(query);
            log.debug("Query "+addr+": " + res.getCount());
            //System.out.println("Going to db for: " + addr);

            for(Map<String, AttributeValue> item : res.getItems())
            {
                if (!item.containsKey(UNSPENT.VALUE))
                {
                    System.out.println(item);
                }
                log.info("  " + item);
                BigInteger value = new BigInteger(item.get(UNSPENT.VALUE).getN());
                total = total.add(value);
            }

            last_key=res.getLastEvaluatedKey();
            query.setExclusiveStartKey(last_key);

        }
        while(last_key != null);


        return total;



    }

    
    public Collection<TransactionOutput> getUnspentOutputs(Address addr)
    {

        LinkedList<TransactionOutput> lst = new LinkedList<TransactionOutput>();

        QueryRequest query = new QueryRequest(table_unspent, new AttributeValue(addr.toString()));
        query.setLimit(100);

        Key last_key=null;

        log.debug("Going to db for: " + addr);
        do
        {
            QueryResult res = dynamo.query(query);
            log.debug("Query "+addr+": " + res.getCount());

            for(Map<String, AttributeValue> item : res.getItems())
            {
                String outpoint=item.get(UNSPENT.OUTPOINT).getS();
                int parse_idx = outpoint.indexOf(":");
                String tx_hash = outpoint.substring(0,parse_idx);
                String idx_str = outpoint.substring(parse_idx+1);
                int idx = Integer.parseInt(idx_str);

                Transaction tx = getTransaction(new Sha256Hash(tx_hash),true);
                if (tx==null)
                {
                    log.error("Unspent with no transaction: " + item);
                    //DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(addr, tx_hash, idx));
                    //dynamo.deleteItem(del);
                    continue;
                    
                }

                TransactionOutput out = tx.getOutputs().get(idx);
                if (out.isAvailableForSpending())
                {
                    lst.add(out);
                }
                else
                {
                    log.error("Erronious unspent: " + item);
                    //DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(addr, tx_hash, idx));
                    //dynamo.deleteItem(del);

                }

            }

            last_key=res.getLastEvaluatedKey();
            query.setExclusiveStartKey(last_key);

        }
        while(last_key != null);


        return lst;
    }
    public Collection<TransactionOutput> getUnspentOutputs(Address addr, BigInteger min_value)
    {
        throw new RuntimeException("Unsupported");
    }

    public void removeSpentBy(Sha256Hash src, int src_idx, Address via_addr, Sha256Hash dst)
    {
        log.info("Removing tx spent " + src + ":" + src_idx);
        //TODO write
        boolean now_unspent=false;

        Map<String, AttributeValue> map = getTransactionMap(src);
        if (map!=null)
        {
            String key = TXgetSpentKey(src_idx);
            if (map.containsKey(key))
            {
                LinkedList<String> old_lst=new LinkedList<String>();
                old_lst.addAll(map.get(key).getSS());

                LinkedList<String> new_lst=new LinkedList<String>();
                for(String s : old_lst)
                {
                    if (!s.equals(dst.toString()))
                    {
                        new_lst.add(s);
                    }
                }
                if (new_lst.size() < old_lst.size())
                {
                    Map<String, AttributeValueUpdate> updates=new TreeMap<String, AttributeValueUpdate>();
                    Map<String, ExpectedAttributeValue> expected=new TreeMap<String, ExpectedAttributeValue>();

                    expected.put(key, new ExpectedAttributeValue(map.get(key)));
                    log.info("Old list: " + old_lst + " new list: " + new_lst);
                    if (new_lst.size() >0)
                    {
                        updates.put(key, new AttributeValueUpdate(new AttributeValue().withSS(dst.toString()), AttributeAction.DELETE));
                    }
                    else
                    {
                        updates.put(key, new AttributeValueUpdate(new AttributeValue().withSS(dst.toString()), AttributeAction.DELETE));
                        now_unspent=true;
                    }
                    UpdateItemRequest req = new UpdateItemRequest(table_tx, TXgetKey(src), updates);
                    //req.setExpected(expected);

                    dynamo.updateItem(req);

                }

            }
        }
        if (now_unspent)
        {   
            Transaction tx = decodeTransaction(map);
            if (tx!=null)
            {
                log.info("Saving unspent: " + via_addr + " " + src + ":" + src_idx);
                saveUnspent(via_addr, src, src_idx, tx.getOutputs().get(src_idx).getValue());
            }
        }

    }

    private void saveSpentBy(Sha256Hash src, long src_idx, Address via_addr, Sha256Hash dst, boolean allow_double_spend)
        throws DoubleSpendException
    {
        //Get rid of the unspent, if it is there
        try
        {
            DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(via_addr, src, src_idx));
            dynamo.deleteItem(del);
        }
        catch(com.amazonaws.services.dynamodb.model.ResourceNotFoundException e)
        {}

        GetItemRequest get = new GetItemRequest(table_tx, TXgetKey(src));
        get.setConsistentRead(consistent_read);
        TreeSet<String> attribs=new TreeSet<String>();
        String spent_key = TXgetSpentKey(src_idx);
        attribs.add( spent_key );

        get.setAttributesToGet(attribs);
        TreeSet<String> spenders=new TreeSet<String>();

        try
        {
            GetItemResult res = dynamo.getItem(get);
            if ((res.getItem() != null) && (res.getItem().containsKey(spent_key)))
            {
                spenders.addAll(res.getItem().get(spent_key).getSS());
            }

            if (spenders.contains(dst.toString())) return;
            if (spenders.size()>0)
            {
                if (!allow_double_spend)
                {
                    throw new DoubleSpendException("Double spend attempt on " + src +":" + src_idx);
                }
            }

            spenders.clear();
            spenders.add(dst.toString());

        
            Map<String, AttributeValueUpdate> m = new TreeMap<String, AttributeValueUpdate>();
            m.put(spent_key, new AttributeValueUpdate(new AttributeValue().withSS(spenders), AttributeAction.ADD));
            UpdateItemRequest update = new UpdateItemRequest(table_tx, TXgetKey(src), m);

            dynamo.updateItem(update);


        }
        catch(com.amazonaws.services.dynamodb.model.ResourceNotFoundException e)
        {
            Map<String, AttributeValue> m = new TreeMap<String, AttributeValue>();
            m.put(spent_key, new AttributeValue(dst.toString()));
            m.put(TX.ID, new AttributeValue(src.toString()));

            PutItemRequest put = new PutItemRequest(table_tx, m);

            Map<String, ExpectedAttributeValue> expected = new TreeMap<String, ExpectedAttributeValue>();
            expected.put(TX.ID, new ExpectedAttributeValue(false));
            put.setExpected(expected);

            dynamo.putItem(put);
        }

    }

    public void saveTx(Transaction tx, boolean allow_double_spend) throws VerificationException, DoubleSpendException
    {
        if (tx.getUpdateTime().getTime()==0)
        {
            tx.setUpdateTime(new java.util.Date());
        }
        while(true)
        {
            try
            {

                saveTxOnce(tx, allow_double_spend);
                return;
            }
            catch(DoubleSpendException e)
            {
                log.warn("Double spend save attempt, removing");
                removeTx(tx);
                throw e;
            }
            catch(Throwable t)
            {
                t.printStackTrace();
                log.error("Unable to save transaction: " + t, t);
            }
            finally
            {
                invalidateCache(tx.getHash());
            }

        }

    }

    private void saveTxOnce(Transaction tx, boolean allow_double_spend) throws VerificationException, DoubleSpendException
    {
        Sha256Hash hash = tx.getHash();
        Map<String, AttributeValue> old_map = getTransactionMap(hash);

        Map<String, AttributeValue> new_map = convertToTransactionMap(tx, old_map);

        for(TransactionInput in : tx.getInputs())
        {
            if (!in.isCoinBase())
            {
                try
                {
                    TransactionOutPoint out = in.getOutpoint();
                    Sha256Hash src_hash = out.getHash();
                    long idx=out.getIndex();
                    saveSpentBy(src_hash, idx, in.getFromAddress(), hash, allow_double_spend);
                }
                catch(com.google.bitcoin.core.ScriptException e)
                {
                    log.warn("Script problem", e);
                }
            }
        }

        //System.out.println(new_map);

        if (old_map == null)
        { //put
            log.info("Putting TX: " + hash + new_map);
            PutItemRequest put = new PutItemRequest(table_tx, new_map);

            Map<String, ExpectedAttributeValue> expected = new TreeMap<String, ExpectedAttributeValue>();
            expected.put(TX.ID, new ExpectedAttributeValue(false));
            put.setExpected(expected);

            dynamo.putItem(put);

        }
        else
        { //update
            
            if (old_map.equals(new_map))
            {
                log.info("Update no changes: " + hash);
                return;
            }

            Map<String, AttributeValueUpdate> updates=new TreeMap<String, AttributeValueUpdate>();

            //Everything is simply copied
            for(String key : new_map.keySet())
            {
                if ((!old_map.containsKey(key)) || (!old_map.get(key).equals(new_map.get(key))))
                {
                    log.info("Updated key: " + key);
                    updates.put(key, new AttributeValueUpdate(new_map.get(key), AttributeAction.PUT));
                }
            }

            //Block list - build this as an add
            TreeSet<String> blocks = new TreeSet<String>();

            if (tx.getAppearsInHashes() != null)
            {
                for(Sha256Hash h : tx.getAppearsInHashes())
                {
                    blocks.add(h.toString());
                }
            }
            if (old_map.containsKey(TX.BLOCK_LIST))
            {
                for(String h : old_map.get(TX.BLOCK_LIST).getSS())
                {
                    blocks.remove(h);
                }
            }

            if (blocks.size() > 0)
            {
                AttributeValue new_blocks = new AttributeValue().withSS(blocks);
                updates.put(TX.BLOCK_LIST, new AttributeValueUpdate(new_blocks, AttributeAction.ADD));
            }


            UpdateItemRequest req = new UpdateItemRequest(table_tx, TXgetKey(hash), updates);
            dynamo.updateItem(req);
            log.info("Update: " + hash + " new blocks: " + blocks);
            
        }

        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            if (!new_map.containsKey(TXgetSpentKey(idx)))
            if (out.isAvailableForSpending())
            {
                try
                {
                    Address addr = out.getScriptPubKey().getToAddress();
                    saveUnspent(addr, hash, idx, out.getValue());
                }
                catch(com.google.bitcoin.core.ScriptException e){}
            }
            idx++;

        }


        
    }


    public void removeTx(Transaction tx)
    {
        log.warn("Removing tx " + tx.getHash()); 

        HashSet<Sha256Hash> downstream=new HashSet<Sha256Hash>();
        Map<String, AttributeValue> tx_map = getTransactionMap(tx.getHash(), false);

        //Remove unspents
        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            try
            {
                Address addr = out.getScriptPubKey().getToAddress();
                DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(addr, tx.getHash(), idx));
                dynamo.deleteItem(del);

                
            }
            catch(ScriptException e){}

            String map_key = TXgetSpentKey(idx);
            if ((tx_map != null) && (tx_map.containsKey(map_key)))
            {
                AttributeValue av = tx_map.get(map_key);
                for(String s : av.getSS())
                {
                    downstream.add(new Sha256Hash(s));
                }

            }

            idx++;
        }

        //Unmark inputs
        for(TransactionInput in : tx.getInputs())
        {
            TransactionOutPoint out = in.getOutpoint();
            
            try
            {
                removeSpentBy(out.getHash(), (int)out.getIndex(), in.getFromAddress(), tx.getHash());
            }
            catch(ScriptException e){}
        }

        //Remove tx
        Map<String,ExpectedAttributeValue> expected = new TreeMap<String,ExpectedAttributeValue>();
        Map<String, AttributeValue> map = convertToTransactionMap(tx, null);

        if (map.get(TX.BLOCK_LIST)!=null)
        {
            expected.put(TX.BLOCK_LIST, new ExpectedAttributeValue(map.get(TX.BLOCK_LIST)) );
        }
        else
        {
            expected.put(TX.BLOCK_LIST, new ExpectedAttributeValue(false));
        }

        DeleteItemRequest req = new DeleteItemRequest(table_tx, TXgetKey(tx.getHash()));
        req.setExpected(expected);
        

        dynamo.deleteItem(req);

        final Semaphore ready=new Semaphore(0);
        int children=downstream.size();

        for(final Sha256Hash hash : downstream)
        {
            executor.execute(new Runnable()
            {
                public void run()
                {
                    try
                    {
                        Transaction tx_down = getTransaction(hash,false);
                        if (tx_down!=null)
                        {
                            log.warn("Removing child transaction " + hash);
    
                            removeTx(tx_down);
                        }
                    }
                    catch(Exception e)
                    {
                        e.printStackTrace();

                    }
                    finally
                    {
                        ready.release(1);
                    }
                }

            });
        }
        try
        {
            ready.acquire(children);
        }
        catch(Exception e)
        {}



        
    }

    private void saveUnspent(Address addr, Sha256Hash tx, long idx, BigInteger value)
    {
        Map<String, AttributeValue> map = new TreeMap<String, AttributeValue>();

        map.put(UNSPENT.ADDRESS, new AttributeValue(addr.toString()));
        map.put(UNSPENT.OUTPOINT, new AttributeValue(UNSPENTgetOutpoint(tx, idx)));
        map.put(UNSPENT.VALUE, new AttributeValue().withN(value.toString()));

        PutItemRequest put = new PutItemRequest(table_unspent, map);

        dynamo.putItem(put);
        invalidateCache(tx);
    }

    private boolean isUnspent(Address addr, Sha256Hash tx, long idx)
    {
        if (stupid_unspent != null)
        {
            String key = "" + addr +":" + tx + ":" + idx;
            return (stupid_unspent.containsKey(key));
        }
        GetItemRequest get = new GetItemRequest(table_unspent, UNSPENTgetKey(addr, tx, idx));
        try
        {
            log.debug("Going to db for unspent: " + addr +" " + tx + ":" + idx);
            GetItemResult res = dynamo.getItem(get);
            if (res.getItem() != null) return true;

        }
        catch(com.amazonaws.services.dynamodb.model.ResourceNotFoundException e)
        {
            return false;
        }
        return false;
    }

    public void enableStupidCache()
    {
        stupid_cache=new HashMap<Sha256Hash, Map<String, AttributeValue>>();
        
    }
    public Set<Transaction> getTransactions(boolean includeDead, boolean includeInactive)
    {
        HashSet<Transaction> set = new HashSet<Transaction>();

            ScanRequest scan = new ScanRequest(table_tx);
            scan.setLimit(1000);
            Key last_key = null;
            do
            {
 
                ScanResult res = dynamo.scan(scan);
                log.info("Tx Query: " + res.getCount());
                for(Map<String, AttributeValue> item : res.getItems())
                {
                    Transaction tx = decodeTransaction(item);
                    if (tx!=null)
                    {
                        set.add(tx);
                        synchronized(tx_cache)
                        {
                            tx_cache.put(tx.getHash(), item);
                        }
                    }
                }

                last_key = res.getLastEvaluatedKey();
                scan.setExclusiveStartKey(last_key);

            }
            while(last_key != null);

        return set;

    }
    public void deleteHopeless(boolean do_delete)
        throws com.google.bitcoin.core.ScriptException
    {
            ScanRequest scan = new ScanRequest(table_tx);
            scan.setLimit(100);
            Key last_key = null;
            do
            {
                ScanResult res = dynamo.scan(scan);
                log.info("Tx Query: " + res.getCount());

                for(Map<String, AttributeValue> item : res.getItems())
                {
                    Transaction tx = decodeTransaction(item);
                    if (tx!=null)
                    if (tx.getConfidence().getConfidenceType()!=ConfidenceType.BUILDING)
                    {
                        long updated=tx.getUpdateTime().getTime();
                        double age=(System.currentTimeMillis() - updated) / 1000.0 / 3600.0;

                        if (updated < System.currentTimeMillis() - HOPELESS_HOURS * 3600L * 1000L)
                        {
                            log.warn("Hopeless: " + tx.getHash() + " age: " + age + " hours");
                            if (do_delete) removeTx(tx);
                        }

                    }
                }

                last_key = res.getLastEvaluatedKey();
                scan.setExclusiveStartKey(last_key);

            }

            while(last_key != null);


    }
    public class CheckStatus
    {
        public int minor=0;
        public int major=0;
        public int tx_count=0;
        public int unspent_count=0;

    }

    public void cacheEverything()
    {
       
        stupid_unspent=new HashMap<String, Map<String, AttributeValue> >();

         { //All transactions
            ScanRequest scan = new ScanRequest(table_tx);
            Key last_key = null;
            scan.setLimit(100);
            do
            {
                ScanResult res = dynamo.scan(scan);
                log.info("Tx Query: " + res.getCount() + " " + res.getConsumedCapacityUnits());

                for(Map<String, AttributeValue> item : res.getItems())
                {
                    stupid_cache.put(new Sha256Hash(item.get(TX.ID).getS()), item);
                }

               //try { Thread.sleep(res.getCount()*4); } catch(Exception e){}

               last_key = res.getLastEvaluatedKey();
               scan.setExclusiveStartKey(last_key);

            }
            while(last_key != null);
        }

        { //All unspent
            ScanRequest scan = new ScanRequest(table_unspent);
            Key last_key = null;
            scan.setLimit(100);
            do
            {
                ScanResult res = dynamo.scan(scan);
                log.info("Unspent Query: " + res.getCount() + " " + res.getConsumedCapacityUnits());

                for(Map<String, AttributeValue> item : res.getItems())
                {
                    String key = item.get(UNSPENT.ADDRESS).getS() + ":" + item.get(UNSPENT.OUTPOINT).getS();
                    stupid_unspent.put(key, item);
                }
               //try { Thread.sleep(res.getCount()*4); } catch(Exception e){}


               last_key = res.getLastEvaluatedKey();
               scan.setExclusiveStartKey(last_key);

            }
            while(last_key != null);
        

        }
 
      

    }

    public void checkConsistency(Sha256Hash hash)
    {
        CheckStatus check_status=new CheckStatus();

        checkTransaction(check_status, getTransactionMap(hash));

        System.out.println("Major: " + check_status.major);
        System.out.println("Minor: " + check_status.minor);
        System.out.println("Transactions: " + check_status.tx_count);
        System.out.println("Unspent: " + check_status.unspent_count);
        
    }
    public void checkConsistency(Collection<Sha256Hash> hashs)
    {
        final CheckStatus check_status=new CheckStatus();

        final Semaphore sem = new Semaphore(0);

        for(final Sha256Hash hash : hashs)
        {
            executor.execute(new Runnable(){
                public void run()
                {
                    CheckStatus c = new CheckStatus();
                    checkTransaction(c, getTransactionMap(hash));

                    synchronized(check_status)
                    {
                        check_status.major += c.major;
                        check_status.minor += c.minor;

                        check_status.tx_count += c.tx_count;

                        check_status.unspent_count += c.unspent_count;
                    }

                    sem.release(1);

                }

            });

        }
        try
        {
        sem.acquire(hashs.size());
        }catch(Exception e){}

        System.out.println("Major: " + check_status.major);
        System.out.println("Minor: " + check_status.minor);
        System.out.println("Transactions: " + check_status.tx_count);
        System.out.println("Unspent: " + check_status.unspent_count);
        
    }


    public void checkConsistency()
    {
        super.checkConsistency();
        cacheEverything();


        //Minor inconsistency (can be caused by interupted processing, partial commits), can't cause double spends
        //Things marked spent where they are not (spending tx does not exist) *
        //Things missing from unspent *
        //Double spent already *

        //Major inconsistency, things that should always be updated first to not happen even in interupted updates
        //may cause double spends
        //Things spending transactions that are not marked as spent (should not happen) *
        //Extra things in unspent (should not happen) *
        //Zero transaction times *
        LinkedList<Map< String, AttributeValue> > lst = new LinkedList<Map< String, AttributeValue> >();
        lst.addAll(stupid_cache.values());

        final CheckStatus check_status=new CheckStatus();

        final Semaphore sem = new Semaphore(0);
        int started=0;

        for(final Map<String, AttributeValue> item : lst)
        {
            started++;
            executor.execute(new Runnable(){

                public void run()
                {
                    CheckStatus c = new CheckStatus();
                    checkTransaction(c, item);

                    synchronized(check_status)
                    {
                        check_status.major += c.major;
                        check_status.minor += c.minor;
                        check_status.tx_count += c.tx_count;
                        check_status.unspent_count += c.unspent_count;
                    }
                    sem.release(1);

                }

            });
        }

        sem.acquireUninterruptibly(started);

        started=0;

        for(final Map<String, AttributeValue> item : stupid_unspent.values())
        {
            started++;
            executor.execute(new Runnable(){

                public void run()
                {
                    CheckStatus c = new CheckStatus();
                    checkUnspent(c, item);

                    synchronized(check_status)
                    {
                        check_status.major += c.major;
                        check_status.minor += c.minor;
                        check_status.tx_count += c.tx_count;
                        check_status.unspent_count += c.unspent_count;
                    }
                    sem.release(1);

                }

            });
            
        }

        sem.acquireUninterruptibly(started);

        System.out.println("Major: " + check_status.major);
        System.out.println("Minor: " + check_status.minor);
        System.out.println("Transactions: " + check_status.tx_count);
        System.out.println("Unspent: " + check_status.unspent_count);

    }

    private void checkUnspent(CheckStatus check_status, Map<String, AttributeValue> item)
    {
            check_status.unspent_count++;

            String outpoint = item.get(UNSPENT.OUTPOINT).getS();
            int idx_colon = outpoint.indexOf(":");
            String tx_hash = outpoint.substring(0,idx_colon);
            int idx = Integer.parseInt(outpoint.substring(idx_colon+1));
            String address_str = item.get(UNSPENT.ADDRESS).getS();
            Address addr ;
            try
            {
                addr = new Address(params, address_str);

            }
            catch(com.google.bitcoin.core.AddressFormatException e)
            {
                log.error("Unspent has invalid address");
                check_status.major++;
                return;
            }

            Transaction tx = getTransaction(new Sha256Hash(tx_hash), true);
            if (tx==null) tx= getTransaction(new Sha256Hash(tx_hash), false);

            if (tx==null)
            {
                check_status.major++;
                log.error("Unspent " + outpoint + " missing tx");

                DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(addr , tx_hash, idx));
                dynamo.deleteItem(del);

            }
            else
            {
                if (!tx.getOutputs().get(idx).isAvailableForSpending())
                {
                    check_status.major++;
                    log.error("Unspent " + outpoint + " appears to be spent");
                    fixTransaction(new Sha256Hash(tx_hash));
                }

            }


    }

    private void checkTransaction(CheckStatus check_status, Map<String, AttributeValue> item)
    {
        if (item==null) return;

        check_status.tx_count++;
        if (stupid_cache != null) 
        {
            stupid_cache.put(new Sha256Hash(item.get(TX.ID).getS()), item);
        }
                    
        Transaction tx = decodeTransaction(item);


        if (tx!=null)
        {
            //too new to bother with
            if (tx.getUpdateTime().getTime() > System.currentTimeMillis() - 60L * 1000L) return;
            if (tx.getUpdateTime().getTime()==0)
            {
                log.error("Transaction " + tx.getHash() + " time zero");
                check_status.major++;
            }
            for(TransactionInput in : tx.getInputs())
            {
                if (!in.isCoinBase())
                {
                    Sha256Hash src_hash = in.getOutpoint().getHash();
                    Map<String, AttributeValue> src_map = getTransactionMap(src_hash, true);
                    if (src_map == null)
                    {
                        //Note: we don't actually need the source transaction to be fully there
                        //but the spent by lines should at least exist
                        log.error("Transaction " + tx.getHash() + " missing source: " + src_hash);
                        check_status.major++;
                        //This will mark the spents
                        fixTransaction(tx.getHash());

                    }
                    else
                    {
                        long idx = in.getOutpoint().getIndex();
                        if (src_map.containsKey(TXgetSpentKey(idx)))
                        {
                            HashSet<String> set = new HashSet<String>();
                            set.addAll(src_map.get(TXgetSpentKey(idx)).getSS());
                            if (!set.contains(tx.getHash().toString()))
                            {
                                check_status.major++;
                                log.error("Transaction " + src_hash + " missing spent: " + idx + " " + tx.getHash());
                                //This will mark the spents
                                fixTransaction(tx.getHash());
                            }
                        }
                        else
                        {
                            check_status.major++;
                            log.error("Transaction " + src_hash + " missing output spent: " + idx);

                            //This will mark the spents
                            fixTransaction(tx.getHash());
                        }

                                

                    }
                }
            }

            int idx=0;
            for(TransactionOutput out : tx.getOutputs())
            {
                boolean in_unspent=false;
                boolean parse_fail=false;
                Address addr=null;
                try
                {
                    addr = out.getScriptPubKey().getToAddress();
                    in_unspent=isUnspent(addr, tx.getHash(), idx);
                }
                catch(ScriptException e)
                {
                    parse_fail=true;
                }

                if (out.isAvailableForSpending())
                { //Should be on unspent

                    if (!in_unspent)
                    if (!parse_fail)
                    {
                        check_status.minor++;
                        log.warn("Transaction " + tx.getHash() + ":" + idx + " missing from unspent");
                        fixTransaction(tx.getHash());
                    }
                  
                                
                }
                else
                { //Spending transactions should exist
                    HashSet<String> spenders = new HashSet<String>();
                    spenders.addAll(item.get(TXgetSpentKey(idx)).getSS());
                    if (spenders.size() > 1)
                    {
                        check_status.minor++;
                        log.warn("Transaction " + tx.getHash() +":" + idx + " already doublespent:" + spenders);
                    }
                    for(String str : spenders)
                    {
                        Transaction spender = getTransaction(new Sha256Hash(str), true);
                        if (spender==null)
                        {
                            check_status.minor++;
                            log.warn("Transction " + tx.getHash() +":" + idx + " spending transaction does not exist: " + str);
                            log.warn("Removing spent by");
                            removeSpentBy(tx.getHash(), idx, addr, new Sha256Hash(str));
                            fixTransaction(tx.getHash());
                        }

                    }
                    if (in_unspent)
                    {

                        check_status.major++;
                        log.error("Transaction " + tx.getHash() + ":" + idx + " in unspent list erroniously");
                        fixTransaction(tx.getHash());
                    }

                }
                            
                idx++;
            }

        }



    }

    public Transaction decodeTransaction(Map<String, AttributeValue> map)
    {
        byte[] byte_data = null;
        if (map.containsKey(TX.BYTES))
        {
            String bytes_string = map.get(TX.BYTES).getS();
            byte_data =  Base64.decodeBase64(bytes_string.getBytes());
        }
        else if (map.containsKey(TX.S3LOC))
        {
            String s3_key = map.get(TX.S3LOC).getS();
            byte_data = s3Get(s3_key);
        }

        if (byte_data == null) return null;

        Transaction tx = null;
        try
        {
            tx = new Transaction(params, byte_data);
        }
        catch(com.google.bitcoin.core.ProtocolException e)
        {
            throw new RuntimeException(e);
        }

        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            String key = TXgetSpentKey(idx);
            AttributeValue val = map.get(key);


            if ((val!=null) && (val.getSS().size() > 0))
            {
                out.markAsSpent(null);
            }
            else
            {
                out.markAsUnspent();
            }
            idx++;

        }
        if (map.containsKey(TX.BLOCK_LIST))
        {

            for(String block : map.get(TX.BLOCK_LIST).getSS())
            {
                tx.addBlockAppearance(new Sha256Hash(block));
            }

        }
        if (map.containsKey(TX.FIRST_SEEN))
        {
            long tm = Long.parseLong(map.get(TX.FIRST_SEEN).getN());
            tx.setUpdateTime(new java.util.Date(tm));
        }
       
        int depth=block_chain_cache.getDepth(tx.getAppearsInHashes());
        if (depth==0)
        {
            if ((tx.getAppearsInHashes() != null) && (tx.getAppearsInHashes().size()>0))
            {
                tx.getConfidence().setConfidenceType(TransactionConfidence.ConfidenceType.NOT_IN_BEST_CHAIN );
                int height = block_chain_cache.getHeight(tx.getAppearsInHashes());
                tx.getConfidence().setAppearedAtChainHeight(height);
            }
            else
            {
                tx.getConfidence().setConfidenceType(TransactionConfidence.ConfidenceType.NOT_SEEN_IN_CHAIN );
            }
        }
        if (depth>0)
        {       
            tx.getConfidence().setConfidenceType(TransactionConfidence.ConfidenceType.BUILDING );
        }

        return tx;


    }
    public Map<String, AttributeValue> convertToTransactionMap(Transaction tx, Map<String, AttributeValue> old_map)
    {
        Map<String, AttributeValue> map = new TreeMap<String, AttributeValue>();
        map.put(TX.ID, new AttributeValue(tx.getHash().toString()));

        long tm = tx.getUpdateTime().getTime();
        if ((old_map!=null) && (old_map.containsKey(TX.FIRST_SEEN)))
        {
            long old_tm = Long.parseLong(old_map.get(TX.FIRST_SEEN).getN());
            tm = Math.min(tm, old_tm);
        }

        map.put(TX.FIRST_SEEN, new AttributeValue().withN("" + tm));

        LinkedList<String> block_list = new LinkedList<String>();
        if (tx.getAppearsInHashes() !=null)
        {
            for(Sha256Hash hash : tx.getAppearsInHashes())
            {
                block_list.add(hash.toString());
            }
        }

        if (block_list.size() > 0)
        {
            map.put(TX.BLOCK_LIST, new AttributeValue().withSS(block_list));
        }

        //Bytes
        byte[] bytes=tx.bitcoinSerialize();
        if (bytes.length < 10240)
        {
            String byte_string = new String(Base64.encodeBase64(bytes));
            map.put(TX.BYTES, new AttributeValue(byte_string));
        }
        else
        {
            String key = "tx/" + tx.getHash().toString();

            s3PutIfAbsent(key, bytes);
            map.put(TX.S3LOC, new AttributeValue(key));
        }



        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            String key = TXgetSpentKey(idx);
            AttributeValue prev = null;
            if (old_map!=null)
            { prev = old_map.get(key); }

            if (prev!=null)
            {
                map.put(key, prev);
            }

            idx++;
        }

        return map;

    }

    private void s3PutIfAbsent(String key, byte[] data)
    {
        try
        {
            if (s3.getObjectMetadata(s3_bucket, key).getContentLength() == data.length) return;
        }
        catch(com.amazonaws.AmazonServiceException e)
        {
            if (e.getStatusCode()!=404) throw e;
        }

        ObjectMetadata omd = new ObjectMetadata();
        omd.setContentLength(data.length);

        PutObjectRequest put = new PutObjectRequest(s3_bucket, key, new ByteArrayInputStream(data), omd);

        s3.putObject(put);
    }
    private byte[] s3Get(String key)
    {

        S3Object obj = s3.getObject(s3_bucket, key);
        int len = (int)obj.getObjectMetadata().getContentLength();
        byte[] bytes=new byte[len];

        DataInputStream din = new DataInputStream(obj.getObjectContent());

        try
        {
            din.readFully(bytes);
            din.close();
            return bytes;
        }
        catch(java.io.IOException e)
        {
            throw new RuntimeException(e);
        }
    }

    public void printDetail(Sha256Hash hash)
    {
        System.out.println("Transaction: " + hash);
        Map<String, AttributeValue> item = getTransactionMap(hash,false);
        if (item==null)
        {
            System.out.println("  null");
            return;
        }
        Transaction tx = decodeTransaction(item);
        if (tx==null)
        {
            System.out.println("  no transaction bytes");
        }
        else
        {
            System.out.println("  UpdateTime: " + tx.getUpdateTime());
            System.out.println("  Confidence: " + tx.getConfidence());
            System.out.println("  Inputs: ");

            int idx=0;
            for(TransactionInput in : tx.getInputs())
            {
                System.out.println("    in:" + idx + " <- " + in.getOutpoint().getHash()+":" + in.getOutpoint().getIndex());
                try
                {
                System.out.println("      " + in.getFromAddress());
                System.out.println("      " + in.getScriptSig().toString());
                System.out.println("      is base: " + in.isCoinBase());

                Script script = in.getScriptSig();
                System.out.println("      to ip: " + script.isSentToIP());
                System.out.println("      sfrom: " + script.getFromAddress());

                System.out.println("      pubkey: " + printBytesInHex(script.getPubKey()));
                System.out.println("      pubkey: " + printBytesInHex(script.getPubKeyHash()));

                System.out.println("      sto: " + script.getToAddress());

                }
                catch(Exception e){e.printStackTrace();}

                idx++;

            }
            System.out.println("  Seen: " + tx.getAppearsInHashes()); 
            if (item.containsKey(TX.S3LOC))
            {
                System.out.println("  S3Loc: " + item.get(TX.S3LOC));
            }
            if (item.containsKey(TX.BYTES))
            {
                System.out.println("  Bytes size: " + item.get(TX.BYTES).getS().length());
                System.out.println("  Base64: " + item.get(TX.BYTES).getS());
            }
        }
        
        System.out.println("  Outputs: ");
        for(int i=0; i<1000; i++)
        {
            String key = TXgetSpentKey(i);
            if (item.containsKey(key))
            {
                System.out.println("    out:" + i + " -> " + item.get(key));
            }
        }

        


        

    }

    private HashSet<Sha256Hash> fixed_tx=new HashSet<Sha256Hash>();

    public void fixTransaction(Sha256Hash tx_id)
    {
        Map<String, AttributeValue> map = getTransactionMap(tx_id, false);

        Transaction tx = decodeTransaction(map);
        synchronized(fixed_tx)
        {
            if (fixed_tx.contains(tx.getHash())) return;

            fixed_tx.add(tx.getHash());
        }
        log.warn("Fix transaction: " + tx_id);
        //printDetail(tx_id);

        for(TransactionInput in : tx.getInputs())
        {
            if (!in.isCoinBase())
            {
                try
                {
                    TransactionOutPoint out = in.getOutpoint();
                    Sha256Hash src_hash = out.getHash();
                    long idx=out.getIndex();
                    saveSpentBy(src_hash, idx, in.getFromAddress(), tx.getHash(), true);
                }
                catch(com.google.bitcoin.core.ScriptException e)
                {
                    log.warn("Script problem", e);
                }
                catch(com.google.bitcoin.db.DoubleSpendException e)
                {
                    //Not that the double spend is surprising, but the excption is

                    log.error("Unxpected double spend exception", e);
                }
            }
        }



        int idx=0;
        for(TransactionOutput out : tx.getOutputs())
        {
            if (out.isAvailableForSpending())
            {
                try
                {
                    Address addr = out.getScriptPubKey().getToAddress();
                    saveUnspent(addr,tx.getHash(), idx, out.getValue());
                }
                catch(com.google.bitcoin.core.ScriptException e){}
            }
            else
            {

                try
                {
                    Address addr = out.getScriptPubKey().getToAddress();
                    DeleteItemRequest del = new DeleteItemRequest(table_unspent, UNSPENTgetKey(addr, tx.getHash(), idx));
                    dynamo.deleteItem(del);

                }
                catch(com.amazonaws.services.dynamodb.model.ResourceNotFoundException e){}
                catch(com.google.bitcoin.core.ScriptException e){}
            }
            idx++;

        }


       
    }
    public static String printBytesInHex(byte[] d)
    {
        StringBuffer s=new StringBuffer(d.length*2);
        BigInteger bi=new BigInteger(1,d);
        s.append(bi.toString(16));
        while(s.length() < d.length*2)
        {
            s.insert(0, '0');
        }
        return s.toString();

    }

}

